package org.example;

import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.http.codec.multipart.Part;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.server.ResponseStatusException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.UUID;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

@RestController
@RequestMapping("/advanced")
public class AdvancedReactiveController {

    // 1. 流式大文件上传（修复类型问题）
    @PostMapping(value = "/upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<Map<String, String>> handleFileUpload( // 改为具体类型 Map<String, String>
                                                       @RequestPart("file") FilePart filePart,
                                                       @RequestPart(value = "metadata", required = false) Mono<String> metadataMono) {

        String filename = "upload_" + UUID.randomUUID() + "_" + filePart.filename();
        Path path = Paths.get(filename);

        return Mono.using(
                () -> AsynchronousFileChannel.open(
                        path, StandardOpenOption.WRITE, StandardOpenOption.CREATE),
                channel -> DataBufferUtils.write(filePart.content(), channel)
                        .publishOn(Schedulers.boundedElastic())
                        .then(metadataMono != null ?
                                metadataMono.flatMap(meta -> saveMetadata(filename, meta)) :
                                Mono.just("No metadata"))
                        .thenReturn(Collections.singletonMap("status", "success")),
                channel -> {
                    try {
                        channel.close();
                    } catch (IOException ignored) {
                    }
                }
        ).onErrorResume(e -> Mono.just(Collections.singletonMap("error", e.getMessage())));
    }

    private Mono<String> saveMetadata(String filename, String metadata) {
        return Mono.delay(Duration.ofMillis(200))
                .thenReturn("Metadata saved for " + filename);
    }

    // 2. 文件下载（支持断点续传）
    @GetMapping(value = "/download/{filename}", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
    public Mono<ResponseEntity<Flux<DataBuffer>>> downloadFile(
            @PathVariable String filename,
            @RequestHeader HttpHeaders headers) {

        Path path = Paths.get(filename);

        return Mono.fromCallable(() -> Files.exists(path))
                .flatMap(exists -> exists ?
                        Mono.just(path) :
                        Mono.error(new ResponseStatusException(HttpStatus.NOT_FOUND, "File not found")))
                .flatMap(filePath -> {
                    try {
                        long fileSize = Files.size(filePath);

                        // 处理范围请求
                        if (headers.getRange().size() > 0) {
                            long start = headers.getRange().get(0).getRangeStart(fileSize);
                            long end = headers.getRange().get(0).getRangeEnd(fileSize);
                            long rangeLength = end - start + 1;

                            Flux<DataBuffer> dataFlux = DataBufferUtils.readAsynchronousFileChannel(
                                    () -> AsynchronousFileChannel.open(filePath, StandardOpenOption.READ),
                                    new DefaultDataBufferFactory(), 4096);

                            return Mono.just(ResponseEntity.status(HttpStatus.PARTIAL_CONTENT)
                                    .header(HttpHeaders.CONTENT_RANGE, "bytes " + start + "-" + end + "/" + fileSize)
                                    .contentLength(rangeLength)
                                    .body(dataFlux));
                        }

                        // 完整文件下载
                        Flux<DataBuffer> dataFlux = DataBufferUtils.readAsynchronousFileChannel(
                                () -> AsynchronousFileChannel.open(filePath, StandardOpenOption.READ),
                                new DefaultDataBufferFactory(), 4096);

                        return Mono.just(ResponseEntity.ok()
                                .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"" + filename + "\"")
                                .contentLength(fileSize)
                                .body(dataFlux));
                    } catch (IOException e) {
                        return Mono.error(new ResponseStatusException(
                                HttpStatus.INTERNAL_SERVER_ERROR, "Error reading file", e));
                    }
                });
    }

    // 3. 多文件并行上传（修复类型问题）
    @PostMapping(value = "/bulk-upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Flux<Map<String, String>> handleBulkUpload( // 改为具体类型 Map<String, String>
                                                       @RequestBody Flux<Part> parts) {

        return parts
                .ofType(FilePart.class)
                .flatMap(filePart -> {
                    String filename = "bulk_" + System.currentTimeMillis() + "_" + filePart.filename();
                    Path path = Paths.get(filename);

                    return DataBufferUtils.write(filePart.content(), path, StandardOpenOption.CREATE)
                            .then(Mono.just(Map.of(
                                    "filename", filename,
                                    "status", "success")))
                            .onErrorResume(e -> Mono.just(Map.of(
                                    "filename", filename,
                                    "status", "failed",
                                    "error", e.getMessage())));
                }, 4);
    }

    // 4. 动态生成ZIP文件流
    @GetMapping(value = "/zip-download", produces = "application/zip")
    public Mono<ResponseEntity<Resource>> generateZipStream() {
        return Mono.fromCallable(() -> {
                    ByteArrayOutputStream baos = new ByteArrayOutputStream();
                    try (ZipOutputStream zos = new ZipOutputStream(baos)) {
                        zos.putNextEntry(new ZipEntry("file1.txt"));
                        zos.write("Content of file1".getBytes());
                        zos.closeEntry();

                        zos.putNextEntry(new ZipEntry("data.json"));
                        zos.write("{\"key\":\"value\"}".getBytes());
                        zos.closeEntry();

                        zos.finish();
                        return new ByteArrayResource(baos.toByteArray());
                    }
                }).subscribeOn(Schedulers.boundedElastic())
                .map(resource -> ResponseEntity.ok()
                        .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"download.zip\"")
                        .contentLength(resource.contentLength())
                        .body(resource));
    }

    // 5. 服务器推送事件（SSE）
    @GetMapping(value = "/live-updates", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> liveUpdates(@RequestParam String userId) {
        return Flux.interval(Duration.ofSeconds(1))
                .map(tick -> "data: Event#" + tick + " for user " + userId + "\n\n")
                .take(10);
    }

    // 6. 流式CSV处理（简化类型）
    @PostMapping(value = "/process-csv", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Flux<Map<String, String>> streamProcessCsv(@RequestPart("file") FilePart filePart) {
        return filePart.content()
                .map(dataBuffer -> {
                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
                    dataBuffer.read(bytes);
                    DataBufferUtils.release(dataBuffer);
                    return new String(bytes);
                })
                .bufferUntil(content -> content.contains("\n"))
                .flatMap(lines -> Flux.fromArray(String.join("", lines).split("\n"))
                        .filter(line -> !line.trim().isEmpty())
                        .map(line -> {
                            String[] fields = line.split(",");
                            if (fields.length >= 3) {
                                return Map.of(
                                        "id", fields[0],
                                        "name", fields[1],
                                        "value", fields[2]
                                );
                            }
                            return Map.of("error", "Invalid line: " + line);
                        }));
    }

    // 7. 响应式图片处理（返回简单类型）
    @PostMapping(value = "/generate-thumbnail", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<ResponseEntity<Resource>> generateThumbnail(@RequestPart("image") FilePart imagePart) {
        return imagePart.content()
                .reduce(new ByteArrayOutputStream(), (output, buffer) -> {
                    byte[] bytes = new byte[buffer.readableByteCount()];
                    buffer.read(bytes);
                    try {
                        output.write(bytes);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    } finally {
                        DataBufferUtils.release(buffer);
                    }
                    return output;
                })
                .map(ByteArrayOutputStream::toByteArray)
                .map(bytes -> {
                    // 实际应用中这里应使用图像处理库
                    String thumbContent = "Thumbnail for: " + imagePart.filename();
                    return new ByteArrayResource(thumbContent.getBytes());
                })
                .map(resource -> ResponseEntity.ok()
                        .header(HttpHeaders.CONTENT_DISPOSITION,
                                "attachment; filename=\"thumb_" + imagePart.filename() + "\"")
                        .contentLength(resource.contentLength())
                        .body(resource));
    }
}